package demo.db.elasticSearch.highClient;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.settings.ClusterGetSettingsRequest;
import org.elasticsearch.action.admin.cluster.settings.ClusterGetSettingsResponse;
import org.elasticsearch.client.ClusterClient;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.Node;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.MainResponse;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * <h1>Java High Level REST Client</h1>
 * <p>
 * 从6.0.0开始加入的，目的是以java面向对象的方式来进行请求、响应处理。 每个API 支持 同步/异步
 * 两种方式，同步方法直接返回一个结果对象。异步的方法以async为后缀，通过listener参数来通知结果。
 * <p>
 * 兼容性说明：依赖 java1.8 和 Elasticsearch core project,请使用与服务端ES版本一致的客户端版本
 */
public class RestHighLevelClientDemo {

    private static Logger logger = LoggerFactory.getLogger(RestHighLevelClientDemo.class);

    private static int connectTimeout = 1000;

    private static int socketTimeout = 30_000;

    private static int connectionRequestTimeout = 500;

    public static int maxConnPerRoute = 10;

    public static int maxConnTotal = 30;

    public static final String host = "192.168.60.98";
    public static final int port = 9200;
    public static final String user = "elastic";
    public static final String pass = "xdcs@xmly";
    public static final String hosts = "192.168.60.98:9200,192.168.60.99:9200";

    /**
     * 获取客户端
     *
     * @return
     */
    public static RestHighLevelClient getClient() {
        return init(hosts);
    }

    /**
     * Initialization
     *
     * @param host
     * @param port
     * @return
     */
    public static RestHighLevelClient newClient(String host, int port) {
        HttpHost httpHost = new HttpHost(host, port, HttpHost.DEFAULT_SCHEME_NAME);
        RestClientBuilder restClientBuilder = RestClient.builder(httpHost);
        return new RestHighLevelClient(restClientBuilder);
    }

    /**
     * Initialization
     *
     * @param host
     * @param port
     * @param user
     * @param pass
     * @return
     */
    public static RestHighLevelClient newClient(String host, int port, String user, String pass) {
        HttpHost httpHost = new HttpHost(host, port, HttpHost.DEFAULT_SCHEME_NAME);

        // 需要用户名和密码的认证
        UsernamePasswordCredentials credentials = new UsernamePasswordCredentials(user, pass);
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, credentials);
        RestClientBuilder restClientBuilder = RestClient.builder(httpHost);
        HttpClientConfigCallback httpClientConfigCallback = new HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        };
        restClientBuilder.setHttpClientConfigCallback(httpClientConfigCallback);
        return new RestHighLevelClient(restClientBuilder);
    }

    /**
     * Initialization
     *
     * @param hosts
     * @param port
     * @param user
     * @param pass
     * @return
     */
    public static RestHighLevelClient newClient(List<String> hosts, int port, String user, String pass) {
        List<HttpHost> httpHostList = hosts.stream()
                .map(host -> new HttpHost(host, port, HttpHost.DEFAULT_SCHEME_NAME))
                .collect(Collectors.toList());

        // 需要用户名和密码的认证
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, pass));

        RestClientBuilder restClientBuilder = RestClient.builder(httpHostList.toArray(new HttpHost[0]));
        HttpClientConfigCallback httpClientConfigCallback = new HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        };
        restClientBuilder.setHttpClientConfigCallback(httpClientConfigCallback);
        return new RestHighLevelClient(restClientBuilder);
    }

    /**
     * 给定集群的多个节点地址，将客户端负载均衡地向这个节点地址集发请求
     *
     * @param hosts
     * @return
     */
    public static RestHighLevelClient init(String hosts) {
        String[] hostArray = hosts.split(",");
        HttpHost[] httpHosts = new HttpHost[hostArray.length];
        for (int i = 0; i < hostArray.length; i++) {
            String[] instance = hostArray[i].split(":");
            httpHosts[i] = new HttpHost(instance[0], Integer.parseInt(instance[1]), HttpHost.DEFAULT_SCHEME_NAME);
        }
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, pass));
        RestClientBuilder builder = RestClient.builder(httpHosts)
                .setRequestConfigCallback(requestConfigBuilder -> {
                    requestConfigBuilder.setConnectTimeout(connectTimeout);
                    requestConfigBuilder.setSocketTimeout(socketTimeout);
                    requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeout);
                    return requestConfigBuilder;
                }).setHttpClientConfigCallback(httpClientBuilder -> {
                    httpClientBuilder.setMaxConnTotal(maxConnTotal);
                    httpClientBuilder.setMaxConnPerRoute(maxConnPerRoute);
                    httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    return httpClientBuilder;
                });
        return new RestHighLevelClient(builder);
    }

    /**
     * info
     *
     * @throws IOException
     */
    @Test
    public void testInfo() throws IOException {
        RestHighLevelClient client = init(hosts);
        MainResponse mainResponse = client.info(RequestOptions.DEFAULT);
        System.err.println(JSON.toJSONString(mainResponse, true));

        boolean pong = client.ping(RequestOptions.DEFAULT);
        System.err.println("ping : " + pong);

        ClusterHealthResponse healthResponse = client.cluster()
                .health(new ClusterHealthRequest(), RequestOptions.DEFAULT);
        System.err.println(JSON.toJSONString(healthResponse, true));
        client.close();
    }

    /**
     * health
     *
     * @throws IOException
     */
    @Test
    public void testCluster() throws IOException {
        RestHighLevelClient client = init(hosts);
        ClusterClient clusterClient = client.cluster();
        // http://192.168.60.99:9200/_cluster/health
        ClusterHealthResponse healthResponse = clusterClient.health(new ClusterHealthRequest(), RequestOptions.DEFAULT);
        System.err.println("health : " + JSON.toJSONString(healthResponse, true));
        ClusterHealthStatus status = healthResponse.getStatus();
        System.err.println(healthResponse.getClusterName() + ": " + status.name());

        ClusterGetSettingsResponse settings = clusterClient.getSettings(new ClusterGetSettingsRequest(), RequestOptions.DEFAULT);
        System.err.println("DefaultSettings: " + settings.getDefaultSettings());
        System.err.println("PersistentSettings: " + settings.getPersistentSettings());
        System.err.println("TransientSettings: " + settings.getTransientSettings());

        client.close();
    }


    /**
     * health
     *
     * @throws IOException
     */
    @Test
    public void testCluster2() throws IOException {
        RestHighLevelClient client = init(hosts);
        RestClient restClient = client.getLowLevelClient();
        List<Node> nodes = restClient.getNodes();
        nodes.forEach(System.err::println);
        client.close();
    }

    /**
     * 带用户名密码校验
     *
     * @throws IOException
     */
    @Test
    public void testAuthorization() throws IOException {
        HttpHost httpHost = new HttpHost(host, port, HttpHost.DEFAULT_SCHEME_NAME);

        // 需要用户名和密码的认证
        UsernamePasswordCredentials credentials = new UsernamePasswordCredentials(user, pass);
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, credentials);
        RestClientBuilder restClientBuilder = RestClient.builder(httpHost);
        HttpClientConfigCallback httpClientConfigCallback = new HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        };
        restClientBuilder.setHttpClientConfigCallback(httpClientConfigCallback);
        RestHighLevelClient client = new RestHighLevelClient(restClientBuilder);
        MainResponse mainResponse = client.info(RequestOptions.DEFAULT);
        System.out.println(JSON.toJSONString(mainResponse, SerializerFeature.PrettyFormat));

        boolean ping = client.ping(RequestOptions.DEFAULT);
        System.out.println("ping :" + ping);
        client.close();
    }

    public void testElasticsearchClient() {
        ElasticsearchClient client = null;
    }


}
